package com.feidee.data.email.dao.impl;

import com.feidee.data.email.dao.PutDao;
import com.feidee.data.email.model.EConstant;
import com.feidee.data.report.exception.DataDaoException;
import com.feidee.data.report.util.JdbcManager;
import org.springframework.stereotype.Repository;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

@Repository
public class PutDaoImpl implements PutDao {


  /*@Resource(name = "mysqlDataSource")
  private DataSource dataSource;*/

  private DataSource sensitiveDataSource;

  @Override
  public void updateDate(List<Map<String, String>> datas, String sink_table, String sink_keys)
      throws DataDaoException {
    Connection conn = null;
    PreparedStatement ps = null;
    if (datas == null || datas.size() == 0) {
      return;
    }
    Set<String> keySet = datas.get(0).keySet();
    String[] keys = sink_keys.split(",");
    List<String> non_keyList = new ArrayList<String>();
    for (String key : keySet) {
      boolean isKey = false;
      for (String k : keys) {
        if (key.equals(k)) {
          isKey = true;
          break;
        }
      }
      if (!isKey) {
        non_keyList.add(key);
      }
    }
    String[] non_keys = new String[non_keyList.size()];
    non_keys = non_keyList.toArray(non_keys);
    StringBuilder sqlBuilder = new StringBuilder("update " + sink_table + " set ");
    for (String non_key : non_keys) {
      sqlBuilder.append(non_key + "=?,");
    }

    sqlBuilder.deleteCharAt(sqlBuilder.length() - 1);
    sqlBuilder.append(" where ");
    for (String key : keys) {
      sqlBuilder.append(key + "=? and ");
    }
    sqlBuilder.append("1=1");
    String sql = sqlBuilder.toString();
    try {
      conn = sensitiveDataSource.getConnection();
      conn.setAutoCommit(false);
      ps = conn.prepareStatement(sql);
      int count = 0;
      for (Map<String, String> data : datas) {

        int i = 1;
        for (String non_key : non_keys) {
          ps.setString(i++, data.get(non_key));
        }
        for (String key : keys) {
          ps.setString(i++, data.get(key));
        }
        ps.addBatch();
        count++;
        if (count % 200 == 0) {
          ps.executeBatch();
          conn.commit();
        }
      }
      ps.executeBatch();
      conn.commit();

    } catch (Exception e) {
      e.printStackTrace();
      throw new DataDaoException(e.getMessage(), e);
    } finally {
      JdbcManager.destroy(conn, ps, null);
    }
  }

  @Override
  public void replaceDateWithVertical(List<Map<String, String>> datas, String sink_table,
                                      String sink_keys) throws DataDaoException {
    Connection conn = null;
    PreparedStatement ps = null;

    StringBuilder sqlBuilder = new StringBuilder("replace into " + sink_table + "(");
    String[] keys = sink_keys.split(",");
    for (String colum : keys) {
      sqlBuilder.append(colum + ",");
    }
    sqlBuilder.append(EConstant.MYSQL_DATA_QUOTA + "," + EConstant.MYSQL_DATA_VALUE + ")VALUES (");
    for (int i = 0; i < keys.length; i++) {
      sqlBuilder.append("?,");
    }
    sqlBuilder.append("?,?)");
    String sql = sqlBuilder.toString();
    try {
      conn = sensitiveDataSource.getConnection();
      conn.setAutoCommit(false);
      ps = conn.prepareStatement(sql);
      int count = 0;
      for (Map<String, String> data : datas) {
        for (Map.Entry<String, String> entry : data.entrySet()) {
          boolean is_key = false;
          for (String key : keys) {
            if (entry.getKey().equals(key)) {
              is_key = true;
              break;
            }
          }
          if (!is_key) {
            int i = 0;
            for (; i < keys.length; i++) {
              ps.setString(i + 1, data.get(keys[i]));
            }
            ps.setString(i + 1, entry.getKey());
            ps.setString(i + 2, entry.getValue());
            count++;
            ps.addBatch();
            if (count % 200 == 0) {
              ps.executeBatch();
              conn.commit();
            }
          }
        }
      }
      ps.executeBatch();
      conn.commit();
    } catch (Exception e) {
      e.printStackTrace();
      throw new DataDaoException(e.getMessage(), e);
    } finally {
      JdbcManager.destroy(conn, ps, null);
    }
  }

  @Override
  public void replaceDataWithHorizontal(List<Map<String, String>> datas, String sink_table)
      throws DataDaoException {
    Connection conn = null;
    PreparedStatement ps = null;
    if (datas == null || datas.size() == 0) {
      return;
    }
    StringBuilder sqlBuilder = new StringBuilder("replace into " + sink_table + "(");
    Set<String> keySet = datas.get(0).keySet();
    String[] keys = new String[keySet.size()];
    keys = keySet.toArray(keys);

    for (String key : keys) {
      sqlBuilder.append(key + ",");
    }
    sqlBuilder.deleteCharAt(sqlBuilder.length() - 1);
    sqlBuilder.append(")VALUES(");
    for (String key : keys) {
      sqlBuilder.append("?,");
    }
    sqlBuilder.deleteCharAt(sqlBuilder.length() - 1);
    sqlBuilder.append(")");
    String sql = sqlBuilder.toString();
    try {
      conn = sensitiveDataSource.getConnection();
      conn.setAutoCommit(false);
      ps = conn.prepareStatement(sql);
      int count = 0;
      for (Map<String, String> data : datas) {
        int i = 0;
        for (; i < keys.length; i++) {
          ps.setString(i + 1, data.get(keys[i]));
        }
        count++;
        ps.addBatch();
        if (count % 200 == 0) {
          ps.executeBatch();
          conn.commit();
        }
      }
      ps.executeBatch();
      conn.commit();
    } catch (Exception e) {
      e.printStackTrace();
      throw new DataDaoException(e.getMessage(), e);
    } finally {
      JdbcManager.destroy(conn, ps, null);
    }
  }

  public void setSensitiveDataSource(DataSource sensitiveDataSource) {
    this.sensitiveDataSource=sensitiveDataSource;
  }
}
